package com.dongnaoedu.rpc.redis.spring;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.ReflectionUtils;

import com.alibaba.fastjson.JSONObject;
import com.dongnaoedu.rpc.redis.Channel;
import com.dongnaoedu.rpc.redis.server.HandlerResult;
import com.dongnaoedu.rpc.redis.server.ServerChannel;

@Component
public class ProviderListener implements MessageListener {

	private final Log log = LogFactory.getLog(getClass());

	private final static String SPLIT_TYPE = Channel.SPLIT;

	private final static ConcurrentHashMap<String, CompletableFuture<HandlerResult>> completableFuture = new ConcurrentHashMap<>();

	private final static ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
	
	private final static int ok = 0;

	private final static int error = -1;

	@Autowired
	StringRedisTemplate stringRedisTemplate;

	@Override
	public void onMessage(Message message, byte[] channelByte) {
		log.info(new String(message.getBody()));
		final String channel = new String(channelByte);
		JSONObject request = JSONObject.parseObject(new String(message.getBody()));
		// parse
		Object[] args = request.getJSONArray("args").toArray();
		String[] types = channel.split(SPLIT_TYPE);
		String _interface = types[1];
		String _method = types[2];
		String backChannel = request.getString("consumerChannel");
		String messageId = splitMessageId(backChannel);
		// reflect
		submit(channel, messageId, _interface, _method, args);
		// back
		stringRedisTemplate.convertAndSend(backChannel, callResult(channel, messageId));
	}

	private String callResult(String channel, String messageId) {
		String result;
		try {
			result = completableFuture.get(channel).get().toString();
		} catch (InterruptedException | ExecutionException e) {
			log.error("call result fail ", e);
			result = handerResult(messageId, e.getMessage()).toString();
		} finally {
			completableFuture.remove(channel);
		}
		return result;
	}

	private void submit(String channel, String messageId, String _interface, String _method, Object[] args) {
		if (completableFuture.contains(channel)) {
			return;
		} else {
			completableFuture.put(channel, new CompletableFuture<>());
		}
		cachedThreadPool.submit(() -> {
			HandlerResult handerResult = reflectInvoke(messageId, _interface, _method, args);
			completableFuture.get(channel).complete(handerResult);
		});
	}

	private HandlerResult reflectInvoke(String messageId, String _interface, String _method, Object[] args) {
		try {
			Channel channel = new ServerChannel(_interface, _method, args);
			Object target = ApplicationContextHolder.getApplicationContext().getBean(channel.getInterface());
			return handerResult(messageId, ReflectionUtils.invokeMethod(channel.getMethod(), target, args));
		} catch (Exception e) {
			return handerResult(messageId, e);
		}
	}

	private String splitMessageId(String channel) {
		String[] types = channel.split(SPLIT_TYPE);
		return types[types.length - 1];
	}

	/**
	 * All results are processed including all known exceptions.
	 * 
	 * @param result
	 * @return
	 */
	private HandlerResult handerResult(String messageId, Object result) {
		if (result instanceof Exception) {
			Exception ex = (Exception) result;
			log.error("Provider deal exception : ", ex);
			if (result instanceof IllegalArgumentException) {
				// java.lang.IllegalArgumentException: argument type mismatch
			} else {
				// java.lang.RuntimeException:
			}
			return new HandlerResult(error, messageId, ex.getMessage());
		} else {
			log.info("Provider deal result success : " + result);
			if (result instanceof String) {
				return new HandlerResult(ok, messageId, new String((String) result));
				// else if ...
			} else {
				return new HandlerResult(ok, messageId, new String(result.toString()));
			}
		}
	}

}
